#include "HiredisSubscriber.h"

#include "HiredisAsyncContext.h"

#include <event2/event.h>
#include <thread>

#define GLOG_USE_GLOG_EXPORT
#include <glog/logging.h>

HiredisSubscriber::HiredisSubscriber() : m_runflag(false)
{
    m_redis_ctx = nullptr;
}

HiredisSubscriber::~HiredisSubscriber()
{
    if (m_redis_ctx) {
        delete m_redis_ctx;
    }
}

void HiredisSubscriber::onMessage(struct redisAsyncContext *ctx, void *reply, void *privdata)
{
    HiredisSubscriber *pthis = (HiredisSubscriber *)privdata;
    redisReply        *r     = (redisReply *)reply;
    if (r == nullptr) {
        return;
    }

    if (r->type == REDIS_REPLY_ARRAY) {
        if (r->elements == 3) {
            if (strcmp(r->element[0]->str, "subscribe") == 0) {
                LOG(INFO) << "subscribe channel " << r->element[1]->str << " success.";
            } else if (strcmp(r->element[0]->str, "message") == 0) {
                pthis->m_callback(r->element[1]->str, r->element[2]->str);
            }
        } else {
            LOG(ERROR) << "error element size: " << r->elements;
        }
    } else {
        LOG(INFO) << "onMessage type: " << r->type;
    }
}

void HiredisSubscriber::taskProcess()
{
    while (m_runflag) {
        m_redis_ctx->connect();
        for (int i = 0; i < m_channels.size(); i++) {
            std::string command = "SUBSCRIBE " + m_channels[i];
            m_redis_ctx->execute(onMessage, this, command.c_str());
        }
        m_redis_ctx->dispatch();
        std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    }
}

bool HiredisSubscriber::init(const std::string &ip, uint16_t port)
{
    m_redis_ctx = new HiredisAsyncContext(ip, port);
    if (!m_redis_ctx->init()) {
        LOG(ERROR) << "Redis context init failed!";
        return false;
    }
    return true;
}

bool HiredisSubscriber::run(const std::vector<std::string> &&channels, SubscribeCallback cb)
{
    m_runflag  = true;
    m_callback = cb;
    m_channels = channels;
    m_thread   = std::thread(&HiredisSubscriber::taskProcess, this);
    return true;
}

void HiredisSubscriber::stop()
{
    m_runflag = false;
    m_redis_ctx->loopbreak();
    if (m_thread.joinable()) {
        m_thread.join();
    }
}
